Deep Dive into Dependency Injection

4 minute read

Published:

“Don’t call us, we’ll call you!”

依赖注入(Dependency Injection,DI)是一种设计模式,其核心思想是:不要让对象自己创建依赖,而是由外部容器来提供依赖。其作为控制反转(IoC)的一种实现方式,已经成为现代软件开发中不可或缺的设计模式。从Spring框架到Angular,从.NET CoreNode.js,无处不在。

1. 什么是依赖注入

🔄 传统方式 vs 依赖注入

传统方式(紧耦合)

# 传统方式 - 对象自己创建依赖
class OrderService:
    def __init__(self):
        self.payment_service = AlipayPaymentService()  # 硬编码依赖
        self.email_service = SmtpEmailService()        # 难以测试
    
    def create_order(self, order_data):
        # 使用硬编码的服务
        payment_result = self.payment_service.process_payment(order_data['amount'])
        self.email_service.send_confirmation(order_data['email'])
        return payment_result

依赖注入方式(松耦合)

# 依赖注入 - 外部提供依赖
class OrderService:
    def __init__(self, payment_service: PaymentService, email_service: EmailService):
        # 构造函数注入
        self.payment_service = payment_service
        self.email_service = email_service
    
    def create_order(self, order_data):
        # 使用注入的服务
        payment_result = self.payment_service.process_payment(order_data['amount'])
        self.email_service.send_confirmation(order_data['email'])
        return payment_result

2. 为什么需要依赖注入

🎯 解决的核心问题

  1. 降低耦合度 - 对象依赖抽象接口而非具体实现
  2. 增强灵活性 - 运行时可以动态替换不同实现
  3. 促进代码复用 - 组件更加通用化

3. 依赖注入的类型

1. 🏗️ 构造函数注入(Constructor Injection)

class OrderService:
    def __init__(self, payment_service: PaymentService, email_service: EmailService):
        # 构造函数注入 - 推荐方式
        self.payment_service = payment_service
        self.email_service = email_service

优点: ✅ 保证依赖完整性,不可变,易于测试

缺点: ❌ 依赖过多时构造函数参数冗长

2. 💉 属性注入

class OrderService:
    def __init__(self):
        self.payment_service = None
        self.email_service = None
    
    def set_payment_service(self, payment_service: PaymentService):
        self.payment_service = payment_service
    
    def set_email_service(self, email_service: EmailService):
        self.email_service = email_service

适用场景: 可选依赖,需要重新配置的场景


4. 实际代码示例

🛍️ 电商订单系统示例

# 步骤1:定义服务接口
class PaymentService(ABC):
    @abstractmethod
    def process_payment(self, amount: float, order_id: str) -> Dict[str, Any]:
        pass

class EmailService(ABC):
    @abstractmethod
    def send_confirmation(self, email: str, order_id: str) -> None:
        pass

class InventoryService(ABC):
    @abstractmethod
    def check_stock(self, product_id: str, quantity: int) -> bool:
        pass
    
    @abstractmethod
    def reduce_stock(self, product_id: str, quantity: int) -> None:
        pass

# 步骤2:实现具体服务
class AlipayPaymentService(PaymentService):
    def process_payment(self, amount: float, order_id: str) -> Dict[str, Any]:
        logging.info(f"Processing Alipay payment for order: {order_id}, amount: {amount}")
        return {"status": "success", "transaction_id": f"alipay_{order_id}"}

class WechatPaymentService(PaymentService):
    def process_payment(self, amount: float, order_id: str) -> Dict[str, Any]:
        logging.info(f"Processing WeChat payment for order: {order_id}, amount: {amount}")
        return {"status": "success", "transaction_id": f"wechat_{order_id}"}

class SmtpEmailService(EmailService):
    def send_confirmation(self, email: str, order_id: str) -> None:
        logging.info(f"Sending SMTP confirmation email to: {email} for order: {order_id}")

class DatabaseInventoryService(InventoryService):
    def __init__(self):
        self.stock = {"product_1": 100, "product_2": 50}  # 模拟数据库
    
    def check_stock(self, product_id: str, quantity: int) -> bool:
        return self.stock.get(product_id, 0) >= quantity
    
    def reduce_stock(self, product_id: str, quantity: int) -> None:
        if self.check_stock(product_id, quantity):
            self.stock[product_id] -= quantity

# 步骤3:订单服务使用依赖注入
class OrderService:
    def __init__(self, payment_service: PaymentService, 
                 email_service: EmailService, 
                 inventory_service: InventoryService):
        self.payment_service = payment_service
        self.email_service = email_service
        self.inventory_service = inventory_service
    
    def create_order(self, order_request: Dict[str, Any]) -> Dict[str, Any]:
        product_id = order_request['product_id']
        quantity = order_request['quantity']
        amount = order_request['amount']
        order_id = order_request['order_id']
        customer_email = order_request['customer_email']
        
        # 1. 检查库存
        if not self.inventory_service.check_stock(product_id, quantity):
            return {"status": "failure", "message": "库存不足"}
        
        # 2. 处理支付
        payment_result = self.payment_service.process_payment(amount, order_id)
        if payment_result["status"] != "success":
            return {"status": "failure", "message": "支付失败"}
        
        # 3. 减少库存
        self.inventory_service.reduce_stock(product_id, quantity)
        
        # 4. 发送确认邮件
        self.email_service.send_confirmation(customer_email, order_id)
        
        return {"status": "success", "message": "订单创建成功"}

# 使用示例
def main():
    # 可以灵活选择不同的实现
    payment_service = AlipayPaymentService()  # 或 WechatPaymentService()
    email_service = SmtpEmailService()
    inventory_service = DatabaseInventoryService()
    
    # 依赖注入
    order_service = OrderService(payment_service, email_service, inventory_service)
    
    # 创建订单
    order_request = {
        'product_id': 'product_1',
        'quantity': 2,
        'amount': 100.0,
        'order_id': 'order_123',
        'customer_email': 'customer@example.com'
    }
    
    result = order_service.create_order(order_request)
    print(result)

if __name__ == "__main__":
    main()

5. 最佳实践指南

✅ 推荐做法

1. 优先使用构造函数注入

# ✅ 推荐 - 构造函数注入
class OrderService:
    def __init__(self, payment_service: PaymentService):
        self.payment_service = payment_service

# ❌ 避免 - 属性直接赋值
class OrderService:
    def __init__(self):
        self.payment_service = AlipayPaymentService()  # 硬编码

2. 依赖接口而非实现

# ✅ 依赖抽象
def __init__(self, payment_service: PaymentService):
    self.payment_service = payment_service

# ❌ 依赖具体实现  
def __init__(self, alipay_service: AlipayPaymentService):
    self.alipay_service = alipay_service

3. 使用类型提示

# ✅ 使用类型提示,提高代码可读性
class OrderService:
    def __init__(self, 
                 payment_service: PaymentService, 
                 email_service: EmailService) -> None:
        self.payment_service = payment_service
        self.email_service = email_service

❌ 避免的反模式

# ❌ 避免
class OrderService:
    def __init__(self, payment_service: PaymentService):
        self.payment_service = payment_service
        self.payment_service.initialize_connection()  # 不推荐

# ✅ 推荐
class OrderService:
    def __init__(self, payment_service: PaymentService):
        self.payment_service = payment_service
    
    def initialize(self):
        self.payment_service.initialize_connection()  # 单独的初始化方法

6. 常见问题与解决方案

🤔 Q1: 什么时候使用依赖注入?

A: 当一个类需要使用其他类的服务时,特别是:

  • 需要提高可测试性
  • 存在多种实现需要切换
  • 需要降低类之间的耦合度

🤔 Q2: 依赖注入会影响性能吗?

A: 现代DI容器的性能开销很小:

  • 对象创建:仅在启动时创建,运行时几乎无开销
  • 内存占用:DI容器本身内存占用很少

🤔 Q3: 如何在Python中实现简单的DI容器?

# 简单的依赖注入容器
class DIContainer:
    def __init__(self):
        self._services = {}
        self._instances = {}
    
    def register(self, interface, implementation, singleton=True):
        self._services[interface] = (implementation, singleton)
    
    def get(self, interface):
        if interface not in self._services:
            raise ValueError(f"Service {interface.__name__} not registered")
        
        implementation, singleton = self._services[interface]
        
        if singleton:
            if interface not in self._instances:
                self._instances[interface] = implementation()
            return self._instances[interface]
        else:
            return implementation()

# 使用示例
container = DIContainer()
container.register(PaymentService, AlipayPaymentService)
container.register(EmailService, SmtpEmailService)

payment_service = container.get(PaymentService)
email_service = container.get(EmailService)
order_service = OrderService(payment_service, email_service)

🤔 Q4: 如何避免循环依赖?

A:重新设计,提取公共依赖

# 避免循环依赖的设计
class NotificationService:
    def notify_user(self, user_id: str, message: str):
        # 被多个服务共同使用,避免循环依赖
        pass

class OrderService:
    def __init__(self, notification_service: NotificationService):
        self.notification_service = notification_service

class UserService:
    def __init__(self, notification_service: NotificationService):
        self.notification_service = notification_service

7. 总结

依赖注入作为现代软件开发的基础设计模式,为我们带来了:

  • 降低耦合度:组件间松散耦合,易于维护
  • 提高可测试性:轻松进行单元测试和集成测试
  • 增强灵活性:支持运行时配置和动态替换
  • 促进代码复用:组件更加通用和模块化